package com.cn.parallel.task;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 并行框架
 * @param <T>
 */
public class TaskExecuter<T> {
	// 存储任务的执行结果
	private List<Future<T>> futres = new ArrayList<Future<T>>();
	// 条件队列锁,以及线程计数器
	public final Lock lock = new Lock();
	// 线程池
	private ExecutorService pool = null;

	public TaskExecuter() {
		this(1);
	}

	public TaskExecuter(int threadPoolSize) {
		pool = Executors.newFixedThreadPool(threadPoolSize);
	}

	/**
	 * 任务派发
	 * @param task
	 */
	public void fork(AbstractTask<T> task) {
		// 设置同步锁
		task.setLock(lock);
		// 将任务派发给线程池去执行
		futres.add(pool.submit(task));
		// 增加线程数
		synchronized (lock) {
			lock.thread_count++;
		}
	}

	/**
	 * 统计任务结果
	 */
	public List<T> join() {
		synchronized (lock) {
			while (lock.thread_count > 0) {// 检查线程数，如果为0，则表示所有任务处理完成
				try {
					lock.wait();// 如果任务没有全部完成，则挂起。等待完成的任务给予通知
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		List<T> list = new ArrayList<T>();
		// 取出每个任务的处理结果，汇总后返回
		for (Future<T> future : futres) {
			try {
				T result = future.get();// 因为任务都已经完成，这里直接get
				if (result != null) {
					list.add(result);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		return list;
	}
	
	public void shutdown() {
		this.pool.shutdown();
	}
}

class Lock {
	// 线程数
	int thread_count;

}